package com.example.myapplication.rxjava;

import android.os.Bundle;
import android.widget.TextView;

import androidx.annotation.Nullable;
import androidx.appcompat.app.AppCompatActivity;
import androidx.lifecycle.Lifecycle;

import com.example.myapplication.R;
import com.orhanobut.logger.Logger;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.AsyncSubject;
import io.reactivex.subjects.BehaviorSubject;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.ReplaySubject;

/**
 * 类描述:
 * 创建人:   tangchao$
 * 创建时间: 2021/8/22$ 9:20 AM$
 * 联系方式: 419704299@qq.com
 * 修改人:   tangchao$
 * 修改时间: 2021/8/22$ 9:20 AM$
 * 修改备注:  [说明本次修改内容]
 */
public class Rxjava2Activity extends AppCompatActivity {
    Disposable disposable;
    CompositeDisposable compositeDisposable = new CompositeDisposable();
    TextView textView;
    int count = 0;

    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.rxjava2activity);
        textView = findViewById(R.id.textView8);
        disposable = Observable.create(new ObservableOnSubscribe<Object>() {
            @Override
            public void subscribe(ObservableEmitter<Object> e) throws Exception {
                Logger.d("Rxjava2Activity===subscribe: 1");
                Thread.sleep(10000);

                for (int i = 0; i < 4; i++) {
                    e.onNext(i);
                    //网络重连
                    if (i == 2) {
                        e.onError(new Throwable("重连"));
                    }
                }

            }
        })
                .retry(new Predicate<Throwable>() {
                    @Override
                    public boolean test(@NonNull Throwable throwable) throws Exception {
                        if (throwable instanceof Throwable && count++ < 4) {
                            Logger.d("Rxjava2Activity===test:重连次数= " + count);
                            return true;
                        }
                        return false;
                    }
                })
                .map(new Function<Object, Object>() {
                    @Override
                    public Object apply(Object o) throws Exception {
                        Logger.d("Rxjava2Activity===apply: 2 ");

                        return o + "map";
                    }
                }).subscribeOn(Schedulers.io())
                .flatMap(new Function<Object, ObservableSource<String>>() {
                    @Override
                    public ObservableSource<String> apply(Object o) throws Exception {
                        Logger.d("Rxjava2Activity===apply:3 ");

                        return Observable.just(o + " flatMap");
                    }
                })
//                .observeOn(AndroidSchedulers.mainThread()
                .concatMap(new Function<Object, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply(Object o) throws Exception {
                        Logger.d("Rxjava2Activity===apply:4 ");

                        return Observable.just(o + " 再一次发射=flatMap");
                    }
                })
                .compose(new SchedulerTransformer<>())//代码复用
                .compose(RxLifecycle.bindRxLifecycle(this))
                .subscribe(new Consumer<Object>() {
                    @Override
                    public void accept(Object o) throws Exception {
                        Logger.d("Rxjava2Activity===accept: 我收到了=" + o);
                        textView.setText(o.toString());
//                        test1();
                        test2();
                    }
                });
//        compositeDisposable.add(disposable);

        //发射事件转换完之后，在flamap时候再次依次发射出去，2，3，2，3    4，5--4.5
    }

    private void test2() {
//        RxPermissions rxPermissions = new RxPermissions(this); // where this is an Activity or Fragment instance
        Observable fIle = Observable.just("文件");
        Observable net = Observable.just("网络");

        Observable.merge(fIle, net).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Logger.d("Rxjava2Activity===accept:一次一次发送过来的 ");

            }
        });
        Observable.zip(fIle, net, new BiFunction<String, String, String>() {
            @NonNull
            @Override
            public String apply(@NonNull String o, @NonNull String o2) throws Exception {
                return o + o2;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Logger.d("Rxjava2Activity===accept: zip合并之后一起发送过来的");

            }
        });

    }
private void test3(){
//        Observable.s
}

    private void test1() {
//        AsyncSubject subject = AsyncSubject.create();
//        BehaviorSubject subject = BehaviorSubject.create();
//        ReplaySubject subject = ReplaySubject.create();
        PublishSubject subject = PublishSubject.create();//用于做rxbus组件通讯

        subject.onNext("12345");
        subject.subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Logger.d("Rxjava2Activity===accept: ");

            }
        });
        subject.onNext("789");
        subject.onComplete();
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();

//        disposable.dispose();//取消订阅
//        compositeDisposable.dispose();//多个的时候一次性取消
    }
}
